package hn.cch.kafka;

import hn.cch.KafkaApp;
import hn.cch.custom.CustomCallback;
import hn.cch.custom.CustomObject;
import hn.cch.custom.CustomProducerInterceptor;
import hn.cch.custom.CustomSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.Future;

public class ProducerDemo {

    private static final Logger logger = LoggerFactory.getLogger(ProducerRunnable.class);

    private Properties properties = new Properties();

    @Before
    public void before() {
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaApp.server);
        // 序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        // properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, BytesSerializer.class);
        // 分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DefaultPartitioner.class);
        // 自定义分区器
        // properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class);

        // 自定义生成者拦截器：注意class.getName()
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, CustomProducerInterceptor.class.getName());
    }

    @Test
    public void test01() {

        try (KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties)) {
            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                        KafkaApp.topic, "i", String.valueOf(i));
                Future<RecordMetadata> future = kafkaProducer.send(producerRecord);
                RecordMetadata recordMetadata = future.get();
                logger.info("partition={},offset={}", recordMetadata.partition(), recordMetadata.offset());
            }
        } catch (Exception e) {
            logger.info(e.getMessage());
        }

    }

    @Test
    public void test02() {
        try (KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties)) {
            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                        KafkaApp.topic, "i", String.valueOf(i));
                // 自定义回调
                Future<RecordMetadata> future = kafkaProducer.send(producerRecord, new CustomCallback());
                RecordMetadata recordMetadata = future.get();
                logger.info("partition={},offset={}", recordMetadata.partition(), recordMetadata.offset());
            }
        } catch (
                Exception e) {
            logger.info(e.getMessage());
        }

    }

    @Test
    public void test03() {
        // 自定义序列化
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CustomSerializer.class);
        try (KafkaProducer<String, CustomObject> kafkaProducer = new KafkaProducer<>(properties)) {
            for (int i = 0; i < 100; i++) {
                CustomObject customObject = CustomObject.createObject(i);
                ProducerRecord<String, CustomObject> producerRecord = new ProducerRecord<>(
                        KafkaApp.topic, "i", customObject);
                // 自定义回调
                Future<RecordMetadata> future = kafkaProducer.send(producerRecord);
                RecordMetadata recordMetadata = future.get();
                logger.info("partition={},offset={}", recordMetadata.partition(), recordMetadata.offset());
            }
        } catch (
                Exception e) {
            logger.info(e.getMessage());
        }

    }
}
